package streaming

import org.apache.spark.SparkConf
import org.apache.spark.streaming.{Seconds, StreamingContext}

object StatefulWordCount {


  def main(args: Array[String]): Unit = {
    val sc = new SparkConf().setMaster("local[*]").setAppName("StatefulWordCount")

    val ssc = new StreamingContext(sc,Seconds(5))

    // 如果使用stateful的算子，必须要设置checkpoint
    // 生产环境中，checkpoint设置在HDFS某个文件夹中
    ssc.checkpoint(".")

    val lines = ssc.socketTextStream("localhost", 9999)

    val result = lines.flatMap(_.split(" ")).map((_,1))

    val state = result.updateStateByKey[Int](updateFunction _)

    state.print()

    ssc.start()

    ssc.awaitTermination()

  }

  /**
    * 根据当前的数据和以前的数据去做更新
    * @param newValues 当前的数据
    * @param runningCount 以前的数据
    * @return
    */
  def updateFunction(newValues: Seq[Int], runningCount: Option[Int]): Option[Int] = {
    val current = newValues.sum
    val pre = runningCount.getOrElse(0)

    Some(current+pre)
  }
}
